package cn.com.rocketmq.batch;

import org.apache.rocketmq.common.message.Message;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * 消息集合切割
 * 消息大小 = 消息长度 + 主题长度 + 消息自定义属性key长度 + 消息自定义属性val长度 + 20（日志空余）
 * @author xuhaixiang
 * @date 2023-09-10
 */
public class ListSplitter implements Iterator<List<Message>> {

    /**
     * 消息大小限制 1MB
     */
    private static final int SIZE_LIMIT = 10 * 1000;

    /**
     * 消息集合
     */
    private final List<Message> messageList;

    /**
     * 当前索引
     */
    private int currentIndex;

    public ListSplitter(List<Message> messageList) {
        this.messageList = messageList;
    }


    @Override
    public boolean hasNext() {
        return currentIndex < messageList.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (; nextIndex < messageList.size(); nextIndex++) {
            Message message = messageList.get(nextIndex);
            int messageSize = message.getBody().length + message.getTopic().length();
            Map<String, String> properties = message.getProperties();
            for (String key : properties.keySet()) {
                String val = properties.get(key);
                messageSize += key.length() + val.length();
            }
            messageSize += 20;
            totalSize += messageSize;
            if (totalSize > SIZE_LIMIT) {
                nextIndex = nextIndex - 1;
                break;
            }
        }
        List<Message> result = messageList.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return result;
    }
}
